package com.baishancloud.log.analyzer

import com.baishancloud.log.common.util.DimMessageFetch
import com.baishancloud.log.format.scala.Origin
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

import java.util.{Timer, TimerTask}
import scala.collection.mutable

/**
 * 根据获取到的最新域名和id过滤数据，并将id和Origin数据做为二元组输出。<br>
 * 域名半小时更新一次<br>
 * 第一个泛型为key的类型，前面的keyBy对随机数进行分区，以打散所有数据，避免数据倾斜。<br>
 * 第二个泛型为实际的值，二元组。第一个元素为实际数据，第二个元素为随机数，用于keyBy。<br>
 * 第三个泛型为输出值，二元组。第一个元素为domainId，第二个元素为origin值。
 *
 * @param parameterTool 入参
 * @author ziqiang.wang
 * @date 2021/11/10 14:38
 */
class DomainIdFilter(parameterTool: ParameterTool) extends ProcessFunction[Origin, (Int, Origin)] {

  private val LOGGER: Logger = LoggerFactory.getLogger(classOf[DomainIdFilter])
  private val domainId: mutable.Map[String, Int] = mutable.Map[String, Int]()
  private var subTaskIndex: Int = -1
  private var allowOldest: Long = System.currentTimeMillis() / 1000 - 30 * 24 * 60L

  override def open(parameters: Configuration): Unit = {
    subTaskIndex = getRuntimeContext.getIndexOfThisSubtask
    DimMessageFetch.fetchAccountDomainId(parameterTool.get(NeteaseAnalyzer.accounts), domainId)
    LOGGER.info(s"DomainIdFilter${subTaskIndex}初始化完成，domainId表大小为：${domainId.size}")
    val timer: Timer = new Timer("update domainId", true)
    //增加定时器，每半小时获取一次最新域名id信息
    timer.scheduleAtFixedRate(new TimerTask {
      override def run(): Unit = {
        DimMessageFetch.fetchAccountDomainId(parameterTool.get(NeteaseAnalyzer.accounts), domainId)
        allowOldest = System.currentTimeMillis() / 1000 - 30 * 24 * 60L
      }
    }, 1800 * 1000, 1800 * 1000)
  }


  override def processElement(value: Origin, ctx: ProcessFunction[Origin, (Int, Origin)]#Context, out: Collector[(Int, Origin)]): Unit = {
    if (domainId.contains(value.domain()) && value.datetimeOfUTC().toEpochSecond >= allowOldest) {
      out.collect((domainId(value.domain()), value))
    }
  }
}
